package xyz.zhouxf.rpc.serializer.impl;

import io.protostuff.LinkedBuffer;
import io.protostuff.ProtostuffIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;
import xyz.zhouxf.rpc.enumeration.SerializerCode;
import xyz.zhouxf.rpc.serializer.RpcSerializer;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author zhouxf
 * @date 2022/8/20
 */
public class ProtobufSerializer implements RpcSerializer {

    private static final LinkedBuffer BUFFER = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);

    private static final Map<Class<?>, Schema<?>> SCHEMA_CACHE = new ConcurrentHashMap<>();

    @Override
    public <T> byte[] serialize(T object) {

        Class<?> clazz = object.getClass();
        Schema<T> schema = (Schema<T>) getSchema(clazz);
        byte[] data;
        try {
            data = ProtostuffIOUtil.toByteArray(object, schema, BUFFER);
        } finally {
            BUFFER.clear();
        }

        return data;
    }


    @Override
    public <T> T deserialize(byte[] data, Class<T> type) {

        Schema<T> schema = getSchema(type);
        T object = schema.newMessage();
        ProtostuffIOUtil.mergeFrom(data, object, schema);

        return object;
    }

    public static <T> Schema<T> getSchema(Class<T> clazz) {

        Schema<T> schema = (Schema<T>) SCHEMA_CACHE.get(clazz);
        if(!SCHEMA_CACHE.containsKey(clazz)) {
            synchronized (ProtobufSerializer.class) {
                if(!SCHEMA_CACHE.containsKey(clazz)) {
                    schema = RuntimeSchema.getSchema(clazz);
                    SCHEMA_CACHE.put(clazz, schema);
                }
            }
        }

        return schema;
    }

    @Override
    public byte getCode() {
        return SerializerCode.PROTOBUF.getCode();
    }
}
